package com.example.transaction.service;

import lombok.Builder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import javax.sql.DataSource;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author tom
 */
@Service
public class MultiThreadingTransactionManagerThree {

    /**
     * 数据源事务管理器
     */
    private DataSourceTransactionManager dataSourceTransactionManager;

    @Autowired
    public void setUserService(DataSourceTransactionManager dataSourceTransactionManager) {
        this.dataSourceTransactionManager = dataSourceTransactionManager;
    }

    /**
     * 是否提交事务,默认是true,当子线程有异常发生时,设置为false,回滚事务
     */
    private final AtomicBoolean isSubmit = new AtomicBoolean(true);

    public boolean execute(List<Runnable> runnableList) {
        List<TransactionStatus> transactionStatusList = Collections.synchronizedList(new ArrayList<>());
        List<TransactionResource> transactionResourceList = Collections.synchronizedList(new ArrayList<>());
        List<CompletableFuture<?>> completableFutureList = new ArrayList<>(runnableList.size());
        runnableList.forEach(runnable -> completableFutureList.add(CompletableFuture.runAsync(() -> {
            try {
                // 执行业务逻辑
                executeThread(runnable, transactionStatusList, transactionResourceList);
            } catch (Exception exception) {
                exception.printStackTrace();
                // 执行异常,需要回滚
                isSubmit.set(false);
                // 终止其它还未执行的任务
                completableFutureList.forEach(completableFuture -> completableFuture.cancel(true));
            }
        })));
        // 等待子线程全部执行完毕
        try {
            // 阻塞直到所有任务全部执行结束,如果有任务被取消,这里会抛出异常,需要捕获
            CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[]{})).get();
        } catch (Exception exception) {
            exception.printStackTrace();
        }
        // 发生了异常则进行回滚操作,否则提交
        if (!isSubmit.get()) {
            System.out.println("发生异常,全部事务回滚");
            for (int i = 0; i < runnableList.size(); i++) {
                transactionResourceList.get(i).autoWiredTransactionResource();
                dataSourceTransactionManager.rollback(transactionStatusList.get(i));
                transactionResourceList.get(i).removeTransactionResource();
            }
        } else {
            System.out.println("全部事务正常提交");
            for (int i = 0; i < runnableList.size(); i++) {
                transactionResourceList.get(i).autoWiredTransactionResource();
                dataSourceTransactionManager.commit(transactionStatusList.get(i));
                transactionResourceList.get(i).removeTransactionResource();
            }
        }
        // 返回结果,是否执行成功,事务提交即为执行成功,事务回滚即为执行失败
        return isSubmit.get();
    }

    private void executeThread(Runnable runnable, List<TransactionStatus> transactionStatusList, List<TransactionResource> transactionResourceList) {
        System.out.println("子线程: [" + Thread.currentThread().getName() + "]");
        DefaultTransactionDefinition defaultTransactionDefinition = new DefaultTransactionDefinition();
        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(defaultTransactionDefinition);
        // 开启新事务
        transactionStatusList.add(transactionStatus);
        // copy事务资源
        transactionResourceList.add(TransactionResource.copyTransactionResource());
        // 执行业务逻辑
        runnable.run();
    }

    /**
     * 保存当前事务资源,用于线程间的事务资源COPY操作
     * <p>
     * `@Builder`注解是Lombok库提供的一个注解,它可以用于自动生成Builder模式的代码,使用@Builder注解可以简化创建对象实例的过程,并且可以使代码更加清晰和易于维护
     */
    @Builder
    private static class TransactionResource {
        // TransactionSynchronizationManager类内部默认提供了下面六个ThreadLocal属性,分别保存当前线程对应的不同事务资源
        // 保存当前事务关联的资源,默认只会在新建事务的时候保存当前获取到的DataSource和当前事务对应Connection的映射关系
        // 当然这里Connection被包装为了ConnectionHolder
        // 事务结束后默认会移除集合中的DataSource作为key关联的资源记录
        private Map<Object, Object> resources;
        //下面五个属性会在事务结束后被自动清理,无需我们手动清理
        // 事务监听者,在事务执行到某个阶段的过程中,会去回调监听者对应的回调接口(典型观察者模式的应用),默认为空集合
        private Set<TransactionSynchronization> synchronizations;
        // 存放当前事务名字
        private String currentTransactionName;
        // 存放当前事务是否是只读事务
        private Boolean currentTransactionReadOnly;
        // 存放当前事务的隔离级别
        private Integer currentTransactionIsolationLevel;
        // 存放当前事务是否处于激活状态
        private Boolean actualTransactionActive;

        /**
         * 对事务资源进行复制
         *
         * @return TransactionResource
         */
        public static TransactionResource copyTransactionResource() {
            return TransactionResource.builder()
                    //返回的是不可变集合
                    .resources(TransactionSynchronizationManager.getResourceMap())
                    //如果需要注册事务监听者,这里记得修改,我们这里不需要,就采用默认负责,spring事务内部默认也是这个值
                    .synchronizations(new LinkedHashSet<>()).currentTransactionName(TransactionSynchronizationManager.getCurrentTransactionName()).currentTransactionReadOnly(TransactionSynchronizationManager.isCurrentTransactionReadOnly()).currentTransactionIsolationLevel(TransactionSynchronizationManager.getCurrentTransactionIsolationLevel()).actualTransactionActive(TransactionSynchronizationManager.isActualTransactionActive()).build();
        }

        /**
         * 使用
         */
        public void autoWiredTransactionResource() {
            resources.forEach(TransactionSynchronizationManager::bindResource);
            //如果需要注册事务监听者,这里记得修改,我们这里不需要,就采用默认负责,spring事务内部默认也是这个值
            TransactionSynchronizationManager.initSynchronization();
            TransactionSynchronizationManager.setActualTransactionActive(actualTransactionActive);
            TransactionSynchronizationManager.setCurrentTransactionName(currentTransactionName);
            TransactionSynchronizationManager.setCurrentTransactionIsolationLevel(currentTransactionIsolationLevel);
            TransactionSynchronizationManager.setCurrentTransactionReadOnly(currentTransactionReadOnly);
        }

        /**
         * 移除
         */
        public void removeTransactionResource() {
            // 事务结束后默认会移除集合中的DataSource作为key关联的资源记录
            // DataSource如果重复移除,unbindResource时会因为不存在此key关联的事务资源而报错
            resources.keySet().forEach(key -> {
                if (!(key instanceof DataSource)) {
                    TransactionSynchronizationManager.unbindResource(key);
                }
            });
        }
    }
}

                        
原文链接：https://blog.csdn.net/qq_30614345/article/details/134631436